package com.swsc.ai.tasks;

import com.swsc.ai.config.SparkSqlConf;
import com.swsc.ai.enums.TagTypeEnum;
import org.apache.spark.sql.*;

import java.util.ArrayList;

/**
 * @author QF
 * @date 2023/10/23 11:25
 * @describe 用户偏好标签计算
 */
public class UserStockPreferenceTask extends UserPreferenceTemplateTask {

    @Override
    public void createAction(SparkSession session, String actionPath, String wideTablePath) throws RuntimeException {
        Dataset<Row> test = null;
        try {
            test = SparkSqlConf.getDataByCVS(session, "user_id STRING, behive STRING", actionPath);
        } catch (Exception e) {
            throw new RuntimeException("读取今日购买行为异常！");
        }
        test.select("user_id", "behive").createOrReplaceTempView("myaction");
        Dataset<Row> sql = session.sql("select user_id,size(split(behive, '\t')),split(behive, '\t') AS actionList from myaction");
        // 使用flatMap函数进行扁平化操作
        Dataset<Row> flatDataset = sql.select(functions.explode(new Column("actionList")), new Column("user_id"))
                .selectExpr("col AS action", "user_id");
        flatDataset.createOrReplaceTempView("user_action");
        Dataset<Row> sql1 = session.sql("select user_id, split(action,':')[0] as tag," +
                "(CASE\n" +
                "    WHEN split(action,':')[1] = 2 THEN -0.2\n" +
                "    ELSE 1\n" +
                " END) * split(action,':')[2] * split(action,':')[3] as amount, split(action,':')[4] buy_date,'astockcode' tag_type from user_action");
        sql1.createOrReplaceTempView("user_action_data");
        sql1.show();
        //创建宽表视图
        Dataset<Row> wideTable = null;
        try {
            wideTable = SparkSqlConf.getDataByCVS(session, "innercode STRING, astockcode STRING, companycode STRING, provincecode STRING, officezipcode STRING, inducsrccode STRING, regcapital DECIMAL, conceptcode STRING, classcode STRING, subclasscode STRING, pettm STRING,  pb STRING, operatingrevenue STRING, grossprofit STRING, totalprofit STRING, totalassets STRING, sewithoutmi STRING, netoperatecashflow STRING, roe STRING, oryoy STRING, grossprofityoy STRING, opyoy STRING, totalprofityoy STRING, totalassettoboy STRING, netocflowyoy STRING, enddate STRING, firstindustrycode STRING, secondindustrycode STRING, thirdindustrycode STRING, fourthindustrycode STRING", wideTablePath);
        } catch (Exception e) {
            throw new RuntimeException("读取宽表数据异常！");
        }
        wideTable.createOrReplaceTempView("wideTable");
        //创建资产等级码表（根据用户数量平均分为5个级别1-5，等级越高资产越多）
        Dataset<Row> regcaptallevel = session.sql("select astockcode,regcapital,ceil(number/((select count(*) from wideTable)/5))regcaptallevel from (select astockcode,regcapital, ROW_NUMBER() OVER (ORDER BY regcapital) AS number from wideTable)");
        regcaptallevel.createOrReplaceTempView("regcaptallevel");
        ArrayList<Dataset<Row>> datasets = new ArrayList<>();
        //各维度统计
        TagTypeEnum[] values = TagTypeEnum.values();
        for (TagTypeEnum value : values) {
            if (0 == value.getTypeCode()) {
                String name = value.name();
                String aliasName = name;
                String sqlString = "select br.user_id AS user_id,wt." + value.name() + " AS tag,'" + aliasName + "' AS tag_type,br.amount AS amount,br.buy_date buy_date from user_action_data br LEFT JOIN wideTable wt where br.tag = wt.innercode";
                if (name.equals("regcapital")) {
                    aliasName = "regcaptallevel";
                    sqlString = "select br.user_id AS user_id,wt.regcaptallevel AS tag,'" + aliasName + "' AS tag_type,br.amount AS amount,br.buy_date buy_date from user_action_data br LEFT JOIN regcaptallevel wt where br.tag = wt.innercode";
                }
                Dataset<Row> dataset = session.sql(sqlString);

                //处理conceptcode、classcode、subclasscode为分隔类型
                if (name.equals("conceptcode") || name.equals("classcode") || name.equals("subclasscode")) {
                    //扁平化dataset
                    Dataset<Row> rowDataset = dataset.withColumn("tag", functions.split(new Column("tag"), "\\|"));
                    Dataset<Row> tag = rowDataset.select(new Column("user_id"), functions.explode(new Column("tag")).name("tag"), new Column("tag_type"), new Column("amount"), new Column("buy_date"));
                    datasets.add(tag);
                } else {
                    datasets.add(dataset);
                }
            }
            if (2 == value.getTypeCode()) {
                //股票非宽表字段
                Dataset<Row> rowDataset = session.sql("select br.user_id AS user_id, 1 AS tag,'" + value.name() + "' AS tag_type,br.amount AS amount,br.buy_date buy_date from user_action_data br");
                datasets.add(rowDataset);
            }
        }
        Dataset<Row> union = (datasets.get(0))
                .union(datasets.get(1))
                .union(datasets.get(2))
                .union(datasets.get(3))
                .union(datasets.get(4))
                .union(datasets.get(5))
                .union(datasets.get(6))
                .union(datasets.get(7))
                .union(datasets.get(8))
                .union(datasets.get(9));
        union.createOrReplaceTempView("user_action_data");
    }
}
